package com.upcoming.data.storm.reliablesend;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.*;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-17 11:44
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class MySpout implements IRichSpout {

    //收集器
    private SpoutOutputCollector collector;

    private Map<Integer, String> toSend = new HashMap<>();
    private Map<Integer, String> allSend = new HashMap<>();
    private Map<Integer, Integer> failMsg = new HashMap<>();

    private Integer MAX_RETRY = 5;

    public MySpout() {
        MyUtil.outLog2NC(this, "new MySpout()");
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("message"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

    /**
     * 初始化
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {

        toSend = Collections.synchronizedMap(toSend);

        allSend = Collections.synchronizedMap(allSend);

        MyUtil.outLog2NC(this, "SPout.open()");
        this.collector = spoutOutputCollector;

        for (int i=0; i < 10; i++) {
            toSend.put(i, "" + i + ",tom" + "," + (10+i));
            allSend.put(i, "" + i + ",tom" + ","  + (10+i));
        }
    }

    public void close() {

        MyUtil.outLog2NC(this, "SPout.close()");
    }

    public void activate() {
        MyUtil.outLog2NC(this, "SPout.activate()");
    }

    public void deactivate() {
        MyUtil.outLog2NC(this, "SPout.deactivate()");
    }

    /**
     *
     */
    public void nextTuple() {
        if (toSend != null && !toSend.isEmpty()) {
            for (Map.Entry<Integer, String> entry : toSend.entrySet()) {
                //发送带消息Id的tuple
                this.collector.emit(new Values(entry.getValue()), entry.getKey());
            }
            toSend.clear();
        }
    }

    public void ack(Object msgId) {
        Integer index = (Integer) msgId;
        toSend.remove(index);
        failMsg.remove(index);
    }

    public void fail(Object msgId) {
        Integer index = (Integer) msgId;
        Integer count = failMsg.get(index);
        count = count == null ? 0 : count;
        if (count < MAX_RETRY) {
            failMsg.put(index, count+1);
            toSend.put(index, allSend.get(index));
        } else {
            MyUtil.outLog2NC(this, "msg:" + index + " has been reach max retry");
            toSend.remove(index);
        }
    }
}
